package com.atguigu.chapter11;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink02_Table_BaseUse {
    public static void main(String[] args) {

        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        );

        // 获取表执行环境，在这之前要获取流的执行环境作为参数传过来
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 1. 用环境，将流转换成表
        Table table = tEnv.fromDataStream(stream);
        // 输出表的详细信息
        table.printSchema();

        // 2. 对表对象进行查询
        // select id,sum(vc) from t group by id
        Table result = table
                .groupBy($("id"))
//                .aggregate($("vc").sum().as("sum_vc"))
//                .select($("id"),$("sum_vc"));
                .select($("id"),$("vc").sum().as("sum_vc"));  // 可以这样写，直接取的时候再聚合，简单

        // 3. 再把动态表转换成流，注意泛型填的是流里面放的数据类型      因为有更新所以用retract流
        SingleOutputStreamOperator<Row> rowDataStream = tEnv.toRetractStream(result, Row.class)
                .filter(t -> t.f0)
                .map(new MapFunction<Tuple2<Boolean, Row>, Row>() {
                    @Override
                    public Row map(Tuple2<Boolean, Row> value) throws Exception {
                        return value.f1;
                    }
                });

        // 4. 输出结果
        rowDataStream.print();


        // ***** 注意一定要提交，要不然没有输出结果
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
